package com.atguigu.PracticalProject.app.dwd.db;

import com.atguigu.PracticalProject.app.BaseSqlApp;
import com.atguigu.PracticalProject.common.Constant;
import com.atguigu.PracticalProject.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdInteractionComment extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdInteractionComment().init(
                2001,
                2,
                "DwdInteractionComment"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        //1.ods_db
        readOdsDb(tEnv,"DwdInteractionComment");


        //2.获取review_info
        Table reviewInfo = tEnv.sqlQuery("select " +
                "data['id']," +
                "data['user_id'] user_id," +
                "data['course_id'] course_id," +
                "data['review_txt'] review_txt," +
                "data['review_stars'] review_stars," +
                "date_format(data['create_time'],'yyyy-MM-dd') date_id," +
                "data['create_time'] create_time," +
                "ts " +
                "from ods_db " +
                "where `database` = 'edu' " +
                "and `table` = 'review_info' " +
                "and `type` = 'insert' " +
                "");


        tEnv.executeSql("create table dwd_interaction_comment(" +
                "id string," +
                "user_id string," +
                "course_id string," +
                "review_txt string," +
                "review_stars string," +
                "date_id string," +
                "create_time string," +
                "ts bigint" +
                ")"+ SQLUtil.getKafkaSink(Constant.TOPIC_DWD_INTERACTION_COMMENT));

        reviewInfo.executeInsert("dwd_interaction_comment");

    }
}
